package mylab.starters.dag;

import com.google.common.collect.Sets;

import java.util.Arrays;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import lombok.Getter;
import lombok.NonNull;
import mylab.starters.dag.graph.impl.BaseVertex;
import mylab.starters.dag.task.Task;
import mylab.starters.dag.task.TaskExecutor;
import mylab.starters.dag.task.TaskFailStrategy;
import mylab.utils.common.StringUtil;

@Getter
public class DagTask<V> extends BaseVertex implements Task<V> {

    Builder<V> builder;
    TaskExecutor<V> taskExecutor;
    TaskFailStrategy failStrategy;
    V defaultValue;
    TimeUnit timeUnit;
    Long timeout;
    Integer retries;
    Set<Object> parameters = Sets.newHashSet();
    AtomicBoolean executed;

    private DagTask(Builder<V> builder) {
        super(builder.id);
        this.name = StringUtil.isNotEmpty(builder.name) ? builder.name : builder.id;
        this.builder = builder;
        this.taskExecutor = builder.executor;
        this.failStrategy = builder.failStrategy;
        this.defaultValue = builder.defaultValue;
        this.timeUnit = builder.timeUnit != null ? builder.timeUnit : TimeUnit.MILLISECONDS;
        this.timeout = builder.timeout;
        this.retries = builder.retries;
        this.parameters.addAll(builder.parameters);
    }

    @Override
    public AtomicBoolean isExecuted() {
        if (executed == null) {
            executed = new AtomicBoolean(false);
        }
        return executed;
    }

    public static class Builder<V> {
        String id;
        String name;
        TaskExecutor<V> executor;
        TaskFailStrategy failStrategy;
        V defaultValue;
        Set<Object> parameters = Sets.newHashSet();
        TimeUnit timeUnit;
        Long timeout;
        Integer retries;


        public Builder(@NonNull String id) {
            this.id = id;
        }

        public Builder<V> name(String name) {
            this.name = name;
            return this;
        }

        public Builder<V> executor(@NonNull TaskExecutor<V> executor) {
            this.executor = executor;
            return this;
        }

        public Builder<V> failStrategy(TaskFailStrategy failStrategy) {
            this.failStrategy = failStrategy;
            return this;
        }


        public Builder<V> defaultValue(V defaultValue) {
            this.defaultValue = defaultValue;
            return this;
        }

        public Builder<V> addParamters(@NonNull Object... parameters) {
            this.parameters.addAll(Arrays.stream(parameters).toList());
            return this;
        }

        public Builder<V> timeUnit(TimeUnit timeUnit) {
            this.timeUnit = timeUnit;
            return this;
        }

        public Builder<V> timeout(Long timeout) {
            this.timeout = timeout;
            return this;
        }

        public Builder<V> retries(Integer retries) {
            this.retries = retries;
            return this;
        }

        public DagTask<V> build() {
            return new DagTask<>(this);
        }


    }
}
